package com.bw.flinkstreaming.state1.job5;

import com.bw.flinkstreaming.state1.job4.MapStateWithCountAvg;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * （hadoop,1）（hadoop,11）（hadoop,20） 20 +11 + 1 / 3
 *
 *   需求：当接收到的相同 key 的元素个数等于 3 个或者超过 3 个的时候
 *  *  就计算这些元素的 value 的平均值。
 *    计算 keyed stream 中每 3 个元素的 value 的平均值
 */

public class ReducingStateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //组织数据源
        DataStreamSource<Tuple2<Long, Long>> dataStreamSource = environment.fromElements
                (Tuple2.of(5L, 12L), Tuple2.of(5L, 11L), Tuple2.of(6L, 9L), Tuple2.of(5L, 8L), Tuple2.of(6L, 52L), Tuple2.of(6L, 77L), Tuple2.of(5L, 42L));
        SingleOutputStreamOperator<Tuple2<Long, Long>> tuple2SingleOutputStreamOperator = dataStreamSource.keyBy(0).flatMap(new ReducingStateWithCountAvg());
        tuple2SingleOutputStreamOperator.print();
        environment.execute("ReducingStateTest");
    }
}
